package com.lzwk.app.ods.adb;

/**
 * @Author: CC
 * @Date: 2022/1/5 17:57
 */

import com.alibaba.fastjson.JSONObject;

import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Field;
import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Schema;
import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.data.Struct;
import com.ververica.cdc.connectors.shaded.org.apache.kafka.connect.source.SourceRecord;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;



import java.util.List;


public class AdbCustomerDeserialization implements DebeziumDeserializationSchema<String> {

    private String flag;

    public AdbCustomerDeserialization(String flag) {
        this.flag = flag;
    }

    /**
     * 封装的数据格式
     * {
     * "database":"",
     * "tableName":"",
     * "before":{"id":"","tm_name":""....},
     * "after":{"id":"","tm_name":""....},
     * "type":"c u d",
     * //"ts":156456135615
     *
     * adb_class_student op: delete -> false  update&create -> true
     * }
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {

        //获取库名&表名
        String topic = sourceRecord.topic();
        String[] fields = topic.split("\\.");
        String database = fields[1];
        String tableName = fields[2];

        Struct value = (Struct) sourceRecord.value();
        //获取"before"数据
        Struct before = value.getStruct("before");
        JSONObject beforeJson = parseToJSON(before);

        //获取"after"数据
        Struct after = value.getStruct("after");
        JSONObject afterJson = parseToJSON(after);

        //获取操作类型  CREATE UPDATE DELETE
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        String type = operation.toString().toLowerCase();

        if("common".equals(flag)) { //common cdc模式

            JSONObject commomResult = parseForCommon(afterJson);
            collector.collect(commomResult.toJSONString());

        }else if("withop".equals(flag)){ //按照规则将before,after,op数据组成JSON

            JSONObject adbResult = parseForAdbClassStudent(type, beforeJson, afterJson);
            collector.collect(adbResult.toJSONString());

        }else {
            //此处可增加其他CDC特殊需求
        }
    }

    private JSONObject parseForCommon(JSONObject afterJson) {

        JSONObject result = new JSONObject();

        result.put("data",afterJson);

        return result;
    }

    private JSONObject parseForAdbClassStudent(String type,JSONObject beforeJson,JSONObject afterJson) {

        JSONObject result = new JSONObject();
        boolean op = true;

        if ("create".equals(type)  || "update".equals(type) || "read".equals(type)) {
            op = true;
            afterJson.put("op",op);
            result.put("data",afterJson);
        }else if("delete".equals(type)){
            op = false;
            beforeJson.put("op",op);
            result.put("data",beforeJson);
        }else{
            result.put("data",null);
        }
        return result;

    }

    private JSONObject parseToJSON(Struct data) {
        JSONObject resJSON = new JSONObject();
        if (data != null) {
            Schema afterSchema = data.schema();
            List<Field> afterFields = afterSchema.fields();
            for (Field field : afterFields) {
                Object afterValue = data.get(field);
                resJSON.put(field.name(), afterValue);
            }
        }
        return resJSON;
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}
